package com.mjf.gmall.realtime.dwd

import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.mjf.gmall.realtime.bean.OrderInfo
import com.mjf.gmall.realtime.bean.dim.{ProvinceInfo, UserState}
import com.mjf.gmall.realtime.util._
import org.apache.hadoop.conf.Configuration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.phoenix.spark._
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 当日新增付费用户首单分析：订单粒度
 *
 * 技术栈：
 * zookeeper/kafka/redis/hadoop/HBase/phoenix/ElasticSearch
 *
 * phoenix建表语句：
 * create table user_state0105(user_id varchar primary key, if_consumed varchar) SALT_BUCKETS = 3;
 */
object OrderInfoApp {
  def main(args: Array[String]): Unit = {

    // 加载流
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("order_info_app")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))

    val topic = "ods_order_info"
    val groupId = "order_info_group"

    // 从redis读取offset
    val kafkaOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(topic, groupId)

    // 从Kafka消费数据
    var recordInputStream: InputDStream[ConsumerRecord[String, String]] = null
    if (kafkaOffsetMap == null && kafkaOffsetMap.size > 0) {
      // 如果redis中已经有了offset信息
      recordInputStream = MyKafkaUtil.getKafkaStream(topic, ssc, kafkaOffsetMap, groupId)
    } else {
      recordInputStream = MyKafkaUtil.getKafkaStream(topic, ssc, groupId)
    }

    //获得偏移结束点
    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange] // 写在算子外面为启动时执行
    val inputGetOffsetDstream: DStream[ConsumerRecord[String, String]] = recordInputStream.transform {
      rdd =>
        // 获得本批次的偏移量的结束位置，用于更新redis中的偏移量
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 写在算子里面是周期性执行（driver端）
        rdd
    }

    // 将json字符串 转换为 OrderInfo对象，补充时间字段
    val orderInfoDStream: DStream[OrderInfo] = inputGetOffsetDstream.map {
      record =>
        val jsonString: String = record.value()
        val orderInfo: OrderInfo = JSON.parseObject(jsonString, classOf[OrderInfo])

        // create_time : 2021-05-29 22:34:04
        val createTimeArr: Array[String] = orderInfo.create_time.split(" ")
        orderInfo.create_date = createTimeArr(0)
        orderInfo.create_hour = createTimeArr(1).split(":")(0)

        orderInfo
    }

    // 查询HBase中用户状态
    /* 问题：查询次数太多
      orderInfoDStream.map{
      orderInfo =>
        val sql = "select user_id,if_consumed from user_state0105 where user_id = '" + orderInfo.user_id + "'"
        val userStatList: List[JSONObject] = PhoenixUtil.queryList(sql)

        if(userStatList != null && userStatList.size > 0) {
          val userStatJsonObj: JSONObject = userStatList(0)
          val if_consumed: String = userStatJsonObj.getString("IF_CONSUMED")
          if(if_consumed.equals("1")) {
            orderInfo.if_first_order = "0"  // 如果消费过，说明不是首单
          } else {
            orderInfo.if_first_order = "1"
          }
        } else {
          orderInfo.if_first_order = "1"  // 如果查询不到用户的状态，说明是首单
        }

        orderInfo
    }*/

    // 查询HBase中用户状态 优化：按分区查询
    val orderInfoWithFirstFlagDStream: DStream[OrderInfo] = orderInfoDStream.mapPartitions {
      orderInfoIter =>
        val orderInfoList: List[OrderInfo] = orderInfoIter.toList
        if (orderInfoList.size > 0) {
          val userId: String = orderInfoList.map(_.user_id).mkString("','")
          val sql = "select user_id,if_consumed from user_state0105 where user_id in ('" + userId + "')"
          // 每个分区只查询一次
          val userStatList: List[JSONObject] = PhoenixUtil.queryList(sql)
          val userStatMap: Map[String, String] = userStatList.map(
            jsonObj =>
              (jsonObj.getString("USER_ID"), jsonObj.getString("IF_CONSUMED"))
          ).toMap

          for (orderInfo <- orderInfoList) { // 每条数据
            val if_consumed: String = userStatMap.getOrElse(orderInfo.user_id.toString(), null)
            if (if_consumed != null && if_consumed.equals("1")) {
              orderInfo.if_first_order = "0" // 如果消费过，说明不是首单
            } else {
              orderInfo.if_first_order = "1" // 如果查询不到用户的状态 或者 没有消费过，说明是首单
            }
          }
        }

        orderInfoList.toIterator
    }

    // 利用HBase进行查询过滤识别首单，只能进行跨批次的判断
    // 如果新用户在同一批次多次下单，会造成该批次该用户都识别为首单 ---需要进行修正
    // 应该同一批次一个用户只有最早的为首单，其他的单据为非首单
    // 处理办法：1 同一批次 同一用户 2 最早的订单   3 标记首单
    //         1 分组：按用户     2 排序：取最早  3 如果最早的订单被标记为首单，除最早的单据一律改为非首单

    // 将原来的结构转为k-v形式，以便可以使用groupByKey
    val orderInfoWithKeyDStream: DStream[(Long, OrderInfo)] = orderInfoWithFirstFlagDStream.map(orderInfo => (orderInfo.user_id, orderInfo))
    // 按用户进行分组
    val orderInfoGroupByUidDStream: DStream[(Long, Iterable[OrderInfo])] = orderInfoWithKeyDStream.groupByKey()
    // 组内进行排序，如果最早的订单被标记为首单，除最早的单据一律改为非首单
    val orderInfoWithFirstRealFlagDStream: DStream[OrderInfo] = orderInfoGroupByUidDStream.flatMap {
      case (userId, orderInfoIter) => {
        if (orderInfoIter.size > 1) {
          // 排序
          val userOrderInfoSortedList: List[OrderInfo] = orderInfoIter.toList.sortWith((orderInfo1, orderInfo2) => orderInfo1.create_time < orderInfo2.create_time)
          val orderInfoFirst: OrderInfo = userOrderInfoSortedList(0)
          if (orderInfoFirst.if_first_order == "1") {
            for (i <- 1 to userOrderInfoSortedList.size - 1) {
              userOrderInfoSortedList(i).if_first_order = "0"
            }
          }
          userOrderInfoSortedList
        } else {
          orderInfoIter.toList
        }
      }
    }


    /* 待优化，可以节省查询次数
    orderInfoWithFirstRealFlagDStream.mapPartitions{
    orderInfoIter =>
      val orderInfoList: List[OrderInfo] = orderInfoIter.toList
      val provinceId: String = orderInfoList.map(_.province_id).mkString("','")
      val sql = "select id,name,area_code,iso_code from gmall0105_province_info where id in ('" + provinceId + "')"
      // 每个分区只查询一次
      val provinceInfoList: List[JSONObject] = PhoenixUtil.queryList(sql)
      null
}*/

    // 优化：因为传输量小，使用数据的占比大，可以考虑使用广播变量。查询HBase的次数变小（分区越多，效果越明显）
    // 利用driver进行查询，在利用广播变量进行分发
    // transform算子内，拿到的RDD算子外，代码是在Driver端执行的，每个batchInterval执行一次，可以做到动态改变广播变量。
    val orderInfoWithProvinceDStream: DStream[OrderInfo] = orderInfoWithFirstRealFlagDStream.transform {
      rdd =>
        // 在driver端按批次周期性执行
        val sql = "select id,name,area_code,iso_code,iso_3166_2 from gmall0105_province_info"
        val provinceInfoList: List[JSONObject] = PhoenixUtil.queryList(sql)
        val provinceMap: Map[String, ProvinceInfo] = provinceInfoList.map {
          jsonObj =>
            val provinceInfo: ProvinceInfo = ProvinceInfo(
              jsonObj.getString("ID"),
              jsonObj.getString("NAME"),
              jsonObj.getString("AREA_CODE"),
              jsonObj.getString("ISO_CODE"),
              jsonObj.getString("ISO_3166_2")
            )
            (provinceInfo.id, provinceInfo)
        }.toMap
        // 将map进行广播
        val provinceInfoBC: Broadcast[Map[String, ProvinceInfo]] = ssc.sparkContext.broadcast(provinceMap)

        val orderInfoWithProvinceRDD: RDD[OrderInfo] = rdd.map { // 在executor端执行
          orderInfo =>
            val provinceMap: Map[String, ProvinceInfo] = provinceInfoBC.value
            val provinceInfo: ProvinceInfo = provinceMap.getOrElse(orderInfo.province_id.toString, null)
            if (provinceInfo != null) {
              orderInfo.province_name = provinceInfo.name
              orderInfo.province_area_code = provinceInfo.area_code
              orderInfo.province_iso_code = provinceInfo.iso_code
              orderInfo.province_iso_3166_2 = provinceInfo.iso_3166_2
            }
            orderInfo
        }
        orderInfoWithProvinceRDD
    }

    //    orderInfoWithProvinceDStream.print(1000)


    // 用户信息关联
    // 注意：用户信息较多，不适合在driver端使用广播变量
    val orderInfoWithUserDStream: DStream[OrderInfo] = orderInfoWithProvinceDStream.mapPartitions {
      orderInfoIter =>
        val orderList: List[OrderInfo] = orderInfoIter.toList
        if (orderList.size > 0) {
          val userId: String = orderList.map(_.user_id).mkString("','")
          val sql = "select id,user_level,birthday,gender,age_group,gender_name from gmall0105_user_info where id in ('" + userId + "')"
          // 每个分区只查询一次
          val userInfoJSONObjList: List[JSONObject] = PhoenixUtil.queryList(sql)
          val userInfoJSONObjMap: Map[Long, JSONObject] = userInfoJSONObjList.map(
            userJSONObj => (userJSONObj.getLongValue("ID"), userJSONObj)
          ).toMap

          for (orderInfo <- orderList) {
            val userInfoJsonObj: JSONObject = userInfoJSONObjMap.getOrElse(orderInfo.user_id, null)
            if (userInfoJsonObj != null) {
              orderInfo.user_age_group = userInfoJsonObj.getString("AGE_GROUP")
              orderInfo.user_gender = userInfoJsonObj.getString("GENDER_NAME")
            }
          }
        }

        orderList.toIterator
    }

    //    orderInfoWithUserDStream.print(1000)


    orderInfoWithUserDStream.cache() // 多次使用，增加缓存

    // 保存 用户状态--->更新HBase 维护状态
    orderInfoWithUserDStream.foreachRDD {
      rdd =>
        // 把首单的订单更新到用户状态中
        val newConsumedUserRDD: RDD[UserState] = rdd.filter(_.if_first_order == "1").
          map(orderInfo => UserState(orderInfo.user_id.toString, "1"))
        newConsumedUserRDD.saveToPhoenix(
          "USER_STATE0105",
          Seq("USER_ID", "IF_CONSUMED"), // Seq中字段的顺序和rdd中字段的顺序要保持一致
          new Configuration,
          Some("hadoop102,hadoop103,hadoop104:2181")
        )
    }

    // 将最终数据保存到es可视化展示
    orderInfoWithUserDStream.foreachRDD {
      rdd =>
        rdd.foreachPartition {
          orderInfoIter =>
            val orderInfoList: List[OrderInfo] = orderInfoIter.toList
            val orderInfoIdList: List[(String, OrderInfo)] = orderInfoList.map(orderInfo => (orderInfo.id.toString, orderInfo))
            val curDate: String = new SimpleDateFormat("yyyy-MM-dd").format(new Date())
            //            MyEsUtil.bulkDoc(orderInfoIdList, "gmall0105_order_info_" + curDate)

            // 保存到 Kafka ---> 继续加工
            for (orderInfo <- orderInfoList) {
              // new SerializeConfig(true) 解决fastJson和Scala的兼容性问题
              val orderInfoJsonString: String = JSON.toJSONString(orderInfo, new SerializeConfig(true))
              // 用订单号分区，防止join时发生shuffle
              MyKafkaSink.send("dwd_order_info", orderInfo.id.toString, orderInfoJsonString)
            }
        }

        // 数据处理结束，偏移量提交到redis
        OffsetManager.saveOffset(topic, groupId, offsetRanges)
    }

    orderInfoWithUserDStream.print(1000)

    // 保存 订单明细 --->OLAP 即席查询


    ssc.start()
    ssc.awaitTermination()

  }
}
